package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.dto.UpdateArticleMessage;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.StringUtils;

import java.time.Duration;

@EnableBinding(value = ArticleProcess.class) // 添加绑定
public class ArticleListener {


    @Value("${commit.time}") // 聚合时间
    private String commitTime;

    // 指定接收消息的主题
    @StreamListener(value = "article_behavior")
    // 指定发送结果的主题
    @SendTo(value = "article_result")
    public KStream<String, String> process(KStream<String, String> input) {
        //1. 接收到的消息格式为 UpdateArticleMessage
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //1.1 key 是接收消息的key , 暂时不关心
                //1.2 value 是json = {"articleId":1540597913363701761,"type":1,"add":1}
                // 打印接收到的value
                System.out.println("接收到消息:" + value);
                //1.3 解析json, 提取文章id
                //1.4 json解析接收前端传来的操作类型消息 里面的操作类型 0 阅读 1 点赞 2 评论 3 收藏
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                //1.5 拿到文章id
                Long articleId = updateArticleMessage.getArticleId();
                //1.6 构建返回的键值对, 因为操作的都是string 所以 文章id转为string 作为key, value 就是value
                return new KeyValue<>(articleId.toString(), value);
            }
        });
        //2.根据每一篇文章的id来进行分组
        KGroupedStream<String, String> groupByKey = map.groupByKey();
        //3. 统计时间窗口的数据
        TimeWindowedKStream<String, String> windowedBy =
                groupByKey.windowedBy(TimeWindows.of(Duration.ofMillis(Long.parseLong(commitTime))));
        //3.1 进行聚合处理
        //3.2 初始化聚合结果
        Initializer<String> init = new Initializer<String>() {
            @Override
            public String apply() {
                // 在同一个时间窗口内, 第一次收到的消息, 返回的聚合结果
                System.out.println("聚合初始化-------");
                //3.3 第一次聚合没有结果, 直接返回null.
                return null;
            }
        };
        //4. 聚合处理的逻辑
        Aggregator<String, String, String> aggregator = new Aggregator<String, String, String>() {
            //4.1 key 是上面自定义的KeyValue中的key --> 文章id
            //4.2 value 是上面自定义的KeyValue中的value --> json = {"articleId":1540597913363701761,"type":1,"add":1}
            //4.3 aggregate 是在同一个时间窗口内上一次聚合的结果
            @Override
            public String apply(String key, String value, String aggregate) {
                // 打印key
                System.out.println("开始本次消息的处理: " + value);
                // 打印value
                System.out.println("上一次聚合的结果: " + aggregate);
                //4.4 每次接收到消息, 都在这个方法内执行一遍
                //4.5 聚合处理的结果是 ArticleStreamMessage 定义初始的处理结果
                ArticleStreamMessage message = null;
                //4.6 判断上一次聚合的结果是否为空
                if (StringUtils.isEmpty(aggregate)){
                    // 如果是空, 构建新的结果
                    message = new ArticleStreamMessage();
                    // 写入文章id, 也就是上面定义的key
                    message.setArticleId(Long.parseLong(key));
                } else {
                    // 如果不是空, 结果直接从上一次提取
                    message = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }
                //4.7 提取本次接收的消息
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                //4.8 处理本次接收的消息进行更新, 操作类型 0 阅读 1 点赞 2 评论 3 收藏
                switch (updateArticleMessage.getType()){
                    case 0:
                        // 0 阅读, 将message中的阅读量加上本次的增量
                        message.setView(message.getView() + updateArticleMessage.getAdd());
                        // 打印 增加的阅读量
                        System.out.println("阅读量增加: " + updateArticleMessage.getAdd());
                        break;
                    case 1:
                        // 1 点赞 将message中的点赞数量加上本次的增量
                        message.setLike(message.getLike() + updateArticleMessage.getAdd());
                        // 打印 增加的点赞量
                        System.out.println("点赞量增加:" + updateArticleMessage.getAdd());
                        break;
                    case 2:
                        // 2 评论 将message中的评论数量加上本次的增量
                        message.setComment(message.getComment() + updateArticleMessage.getAdd());
                        // 打印 增加的评论量
                        System.out.println("增加的评论量: " + updateArticleMessage.getAdd());
                        break;
                    case 3:
                        // 3 收藏 将message中的收藏数量加上本次的增量
                        message.setCollect(message.getCollect() + updateArticleMessage.getAdd());
                        // 打印 增加的收藏量
                        System.out.println("增加的收藏量: " + updateArticleMessage.getAdd());
                        break;
                }
                //4.9 将本次更新的结果保存到 聚合的 中间结果
                String json = JSON.toJSONString(message);
                // 打印本次聚合的结果
                System.out.println("本次聚合的结果: " + json);
                //4.10 返回聚合结果
                return json;
            }
        };
        //5. 获取聚合的结果 ,init初始化聚合结果, aggregator聚合处理的逻辑
        KTable<Windowed<String>, String> aggregate = windowedBy.aggregate(init,aggregator);
        //6. 转换聚合的结果
        KStream<String, String> stream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                // 打印时间窗口的最终处理结果
                System.out.println("打印时间窗口的最终处理结果:" + value);
                // 返回聚合的键值对结果集
                return new KeyValue<>(key.key(), value);
            }
        });
        //7. 最终发送到结果的value是 ArticleStreamMessage转换成json格式
        return stream;
    }
}
